package com.example.mq.mqclient;

import com.example.mq.common.BinaryTool;
import com.example.mq.common.MqException;
import com.example.mq.common.net.BasicReturns;
import com.example.mq.common.net.Request;
import com.example.mq.common.net.Response;
import com.example.mq.common.net.SubScribeReturns;

import java.io.*;
import java.net.Socket;
import java.net.SocketException;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Connection {

    //管理每一个tcp连接
    private Socket socket = null;

    //使用host把多个channel管理起来<channelId,channel>
    private ConcurrentHashMap<String, Channel> channelMap = new ConcurrentHashMap<>();

    private InputStream inputStream;
    private OutputStream outputStream;
    private DataInputStream dataInputStream;
    private DataOutputStream dataOutputStream;
    
    // 当前连接使用的虚拟主机名
    private String virtualHostName = "default";

    private ExecutorService callBackPool = null;

    //关闭connection的所有资源
    public void close() {
        try {
            callBackPool.shutdownNow();
            inputStream.close();
            outputStream.close();
            socket.close();
        } catch (IOException e) {
            System.out.println("【Connection】关闭资源出问题");
            throw new RuntimeException(e);
        }

    }


    //构造方法初始化
    public Connection(String host, int post) throws IOException {
        this(host, post, "default");
    }
    
    //带虚拟主机名的构造方法
    public Connection(String host, int post, String virtualHostName) throws IOException {
        //1、于broker建立TCP连接
        socket = new Socket(host, post);

        //2、初始化输入输出流
        inputStream = socket.getInputStream();
        outputStream = socket.getOutputStream();
        dataInputStream = new DataInputStream(inputStream);
        dataOutputStream = new DataOutputStream(outputStream);
        
        //3、设置虚拟主机名
        if (virtualHostName != null && !virtualHostName.isEmpty()) {
            this.virtualHostName = virtualHostName;
        }

        //4、创建线程池供接收到消息后的回调函数使用
        callBackPool = Executors.newFixedThreadPool(4);

        Thread t = new Thread(() -> {
            try {
                while (!socket.isClosed()) {
                    Response response = readResponse();
                    BasicReturns basicReturns = (BasicReturns) BinaryTool.fromBytes(response.getPayload());

                    dispatchResponse(response);
                }
            } catch (SocketException e) {
                // 连接正常断开的. 此时这个异常直接忽略.
                System.out.println("[Connection] 连接正常断开!");

            } catch (Exception e) {
                System.out.println("[Connection] 连接异常断开!");
                e.printStackTrace();
            }
        });
        t.start();
    }
    
    // 获取当前连接使用的虚拟主机名
    public String getVirtualHostName() {
        return virtualHostName;
    }


    //根据不同响应，执行不同程序
    public void dispatchResponse(Response response) throws IOException, ClassNotFoundException, MqException {
        if (response.getType() == 0xc) {

            //服务器主动发送消息给客户端
            SubScribeReturns subScribeReturns = (SubScribeReturns) BinaryTool.fromBytes(response.getPayload());

            //根据channelID找到指定channel
            Channel channel = channelMap.get(subScribeReturns.getChannelId());
            if (channel == null) {
                throw new MqException("[Connection] 该消息对应的 channel 在客户端中不存在! channelId=" + channel.getChannelId());
            }


            //执行消息回调
            callBackPool.submit(() -> {
                try {
                    System.out.println("【执行了回调方法！！！！！！！！！！！！！！！！！！！！！！！！】");
                    //执当前信道的回调函数
                    channel.getConsumer().handleDelivery(
                            subScribeReturns.getConsumerTag()
                            , subScribeReturns.getBasicProperties()
                            , subScribeReturns.getBody()
                    );
                } catch (MqException | IOException e) {
                    System.out.println("[Connection dispatchResponse]执行回调失败！");
                    throw new RuntimeException(e);
                }
            });
        } else {//其他常规响应（告诉客户端请求是否执行成功

            //获取响应结果
            BasicReturns basicReturns = (BasicReturns) BinaryTool.fromBytes(response.getPayload());



            //根据响应结果中的channelId找到channel
            Channel channel = channelMap.get(basicReturns.getChannelId());


            if (channel == null) {
                throw new MqException("[Connection] 该消息对应的 channel 在客户端中不存在! channelId=" + channel.getChannelId());
            }

            //把服务器响应信息放到 basicReturnMap中，同时提醒当前对象的线程去获取，把响应返回给客户端
            channel.putReturns(basicReturns);

        }
    }

    //发送请求
    public void writeRequest(Request request) throws IOException {

        dataOutputStream.writeInt(request.getType());
        dataOutputStream.writeInt(request.getLength());
        dataOutputStream.write(request.getPayload());
        dataOutputStream.flush();
        System.out.println("[Connection] 发送请求! type=" + request.getType() + ", length=" + request.getLength());

    }

    //读取响应
    public Response readResponse() throws IOException {
        Response response = new Response();
        response.setType(dataInputStream.readInt());
        response.setLength(dataInputStream.readInt());
        byte[] payload = new byte[response.getLength()];


        int actualLen = dataInputStream.read(payload);
        response.setPayload(payload);




        //长度不一致，一定有问题

        if (actualLen != response.getLength()) {
            throw new IOException("[Connection] 客户端读取响应格式出现异常！");
        }



        return response;
    }


    /*
     * 创建信道*/
    public Channel createChannel() throws IOException, InterruptedException {

        String channelId = "Client-" + UUID.randomUUID().toString();
        Channel channel = new Channel(channelId, this);

        //放入map中，管理每一个connection中都有那些channel
        channelMap.put(channelId, channel);

        //把创建channel的消息告诉服务器
        boolean ok = channel.createChannel();

        if (!ok) {
            // 服务器这里创建失败了!! 整个这次创建 channel 操作不顺利!!
            // 把刚才已经加入 hash 表的键值对, 再删了.
            channelMap.remove(channelId);
            return null;
        }


        return channel;
    }

}
